package org.etlt.extract;

import org.etlt.job.JobContext;
import org.etlt.job.config.xml.AbstractXmlResolver;
import org.etlt.load.ColumnSetting;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class ExtractorSettingXmlResolver extends AbstractXmlResolver<ExtractorSetting> {

    @Override
    public List<ExtractorSetting> resolveBundle(InputStream stream, JobContext context) {
        try {
            Document document = getDocumentBuilder().parse(stream);
            Element root = document.getDocumentElement();
            if (root.getNodeName().equals("extractors")) {
                List<ExtractorSetting> bundleSettings = new ArrayList<>();
                NodeList children = root.getElementsByTagName("extractor");
                for (int i = 0; i < children.getLength(); i++) {
                    Node node = children.item(i);
                    if (node.getNodeType() == Node.ELEMENT_NODE) {
                        bundleSettings.add(resolve((Element) node, context));
                    }
                }
                return bundleSettings;
            } else
                return super.resolveBundle(stream, context);
        } catch (SAXException | IOException e) {
            throw new IllegalStateException("job.xml parse error.", e);
        }
    }

    @Override
    protected void registerParser() {

        registerParser( (setting, node, context) -> {
            Element element = (Element) node;
            String type = element.getAttribute("type");
            String name = element.getAttribute("name");
            switch (type) {
                case ExtractorSetting.DATA_BASE_TYPE:
                    DatabaseExtractSetting setting1 = new DatabaseExtractSetting();
                    setting1.setName(name);
                    setting1.setDatasource(getFirstChildValueByName(element, "datasource"));
                    setting1.setDql(getFirstChildValueByName(element, "dql"));
                    return setting1;
                case ExtractorSetting.BIG_DATA_BASE_TYPE:
                    DatabaseExtractSetting setting2 = new DatabaseExtractSetting();
                    setting2.setName(name);
                    setting2.setParallel(Boolean.parseBoolean(this.getAttributeValue(element, "parallel")));
                    setting2.setDatasource(getFirstChildValueByName(element, "datasource"));
                    setting2.setPkDql(getFirstChildValueByName(element, "pkDql"));
                    setting2.setPkName(getFirstChildAttributeValueByName(element, "pkDql", "pkName"));
                    setting2.setUnion(
                            Boolean.parseBoolean(getFirstChildAttributeValueByName(element, "pkDql", "union")));
                    setting2.setDql(getFirstChildValueByName(element, "dql"));
                    setting2.setDqlByPk(getFirstChildValueByName(element, "dqlByPk"));
                    return setting2;
                case ExtractorSetting.FILE_TYPE:
                    FileExtractSetting setting3 = new FileExtractSetting();
                    setting3.setName(name);
                    setting3.setDataSource(getFirstChildValueByName(element, "datasource"));
                    setting3.setDelim(getFirstChildValueByName(element, "delim"));
                    return setting3;
                default:
                    break;
            }
            throw new UnsupportedOperationException("unsupported extractor type: " + type);
        },"/extractor","/extractors/extractor");

        registerParser( ((setting, node, context) -> {
            Element skip = (Element) node;
            setting.setSkip(Integer.parseInt(skip.getTextContent()));
            return setting;
        }),"/extractor/skip");

        registerParser( ((setting, node, context) -> {
            Element criteria = (Element) node;
            LinkedExtractorSetting setting1 = (LinkedExtractorSetting) setting;
            setting1.setSort(criteria.getAttribute("sorting"));
            String criteriaContent = criteria.getTextContent().trim();
            setting1.setCriteria(Arrays.asList(criteriaContent.split(",")));
            return setting;
        }),"/linkedExtractor/criteria");

        registerParser( ((setting, node, context) -> {
            Element element = (Element) node;
            NodeList children = element.getChildNodes();
            for (int i = 0; i < children.getLength(); i++) {
                Node cNode = children.item(i);
                if (cNode.getNodeType() == Node.ELEMENT_NODE) {
                    parseElementForcedly(setting, (Element) cNode, context);
                }
            }
            return setting;
        }), "/extractor/columns","/linkedExtractor/columns");

        registerParser(((setting, node, context) -> {
            Element element = (Element) node;
            String name = getAttributeValue(element, "name");
            String expression = element.getTextContent();
            ColumnSetting columnSetting = new ColumnSetting(name, expression);
            setting.getColumnSettings().add(columnSetting);
            return setting;
        }),"/extractor/columns/column","/linkedExtractor/columns/column");

        registerParser((setting, node, context) -> {
            LinkedExtractorSetting setting1 = new LinkedExtractorSetting();
            Element element = (Element) node;
            setting1.setName(getAttributeValue(element, "name"));
            setting1.setType(getAttributeValue(element, "type"));
            setting1.setExtractors(getChildValuesByName(element, "innerExtractor"));
            return setting1;
        }, "/linkedExtractor");


    }
}
